package com.partsoft.dits.dsm;



import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import javax.sql.DataSource;

import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import com.partsoft.dits.AbstractConfigurableDescriptorManager;
import com.partsoft.dits.BaseDataIntegrateBatch;
import com.partsoft.dits.DataBroadcaster;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.DataIntegrateBatch;
import com.partsoft.dits.DataIntegrateProcessor;
import com.partsoft.dits.DataTransformer;
import com.partsoft.dits.DescriptorDepends;
import com.partsoft.dits.EdgeDescriptor;
import com.partsoft.dits.EdgeException;
import com.partsoft.dits.MonitorStartMode;
import com.partsoft.dits.TableDataEdgeDescriptor;
import com.partsoft.dits.integrate.TableDataDescriptorRegistry;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.DataAction;
import com.partsoft.dits.message.DataProperty;
import com.partsoft.dits.message.MessageEntity;
import com.partsoft.dits.message.MessageFormat;
import com.partsoft.dits.message.MessageProcessor;
import com.partsoft.dits.utils.DbUtils;
import com.partsoft.dits.utils.ExecutionCallback;
import com.partsoft.dits.utils.JdbcTemplate;
import com.partsoft.dits.utils.ReturnExecutionCallback;

@Deprecated
@SuppressWarnings({"unchecked", "rawtypes"})
public class ContextConfigurableDataSourceMonitorReactor extends
		AbstractConfigurableDescriptorManager<DataSourceMonitorDescriptor>
		implements DataSourceMonitorManager,
		DataIntegrateProcessor,
		ApplicationListener<ApplicationEvent>, 
		ApplicationContextAware, 
		DescriptorDepends<DataSourceMonitorDescriptor> {

	private Set<Class<? extends EdgeDescriptor>> dependsClasses; 
	{
		dependsClasses = new HashSet<Class<? extends EdgeDescriptor>>(1);
		dependsClasses.add(DataSourceDescriptor.class);
	}
	
	private Map<String, DataSourceMonitorTask> monitors = Collections.synchronizedMap(new HashMap<String, DataSourceMonitorTask>(1));
	
	private DsmSQLProviderFactory dsmSQLProviderFactory = DefaultDsmSQLProviderFactory.get();
	
	private String monitorLogTable = DsmOriginalSQLs.Constants.MonitorLogTable;
	
	private DataSourceFactory dataSourceFactory;
	
	private MessageProcessor messageProcessor = MessageProcessor.INSTANCE;
	
	private TableDataDescriptorRegistry<TableDataEdgeDescriptor> dataDescriptorRegistry;
	
	private DataTransformer dataTransformer;
	
	private DataBroadcaster dataBroadcaster;
	
	private String monitorLogNS;
	
	public void setMonitorLogTable(String monitorLogTable) {
		this.monitorLogTable = monitorLogTable;
	}
	
	public void setDsmSQLProviderFactory(DsmSQLProviderFactory dsmSQLProviderFactory) {
		this.dsmSQLProviderFactory = dsmSQLProviderFactory;
	}

	public Class<DataSourceMonitorDescriptor> getSupportClass() {
		return DataSourceMonitorDescriptor.class;
	}
	
	public DataBroadcaster getDataBroadcaster() {
		return dataBroadcaster;
	}

	public void setDataBroadcaster(DataBroadcaster dataBroadcaster) {
		assertNoInit();
		this.dataBroadcaster = dataBroadcaster;
	}

	public void setDataSourceFactory(DataSourceFactory dataSourceFactory) {
		assertNoInit();
		this.dataSourceFactory = dataSourceFactory;
	}
	
	public DataTransformer getDataTransformer() {
		return dataTransformer;
	}

	public void setDataTransformer(DataTransformer dataTransformer) {
		assertNoInit();
		this.dataTransformer = dataTransformer;
	}

	public DataSourceFactory getDataSourceFactory() {
		return dataSourceFactory;
	}
	
	public String getMonitorLogNS() {
		return monitorLogNS;
	}

	public void setMonitorLogNS(String namespace) {
		assertNoInit();
		this.monitorLogNS = namespace;
	}

	public void setDataDescriptorRegistry(TableDataDescriptorRegistry<TableDataEdgeDescriptor> dataDescriptorRegsitry) {
		assertNoInit();
		this.dataDescriptorRegistry = dataDescriptorRegsitry;
	}
	
	public TableDataDescriptorRegistry<TableDataEdgeDescriptor> getDataDescriptorRegistry() {
		return dataDescriptorRegistry;
	}
	
	public void setMessageProcessor(MessageProcessor messageProcessor) {
		assertNoInit();
		this.messageProcessor = messageProcessor;
	}
	
	public MessageProcessor getMessageProcessor() {
		return messageProcessor;
	}
	
	private void createAndStartDataSourceMonitor(DataSourceMonitorDescriptor descriptor) {
		synchronized (this) {
			if (descriptor.getStartMode() == MonitorStartMode.DISABLED) return;
			DataSource ds = getDataSourceFactory().getDataSource(descriptor.getDatasourceName());
			DataSourceMonitorContext ctx = new DataSourceMonitorContextImpl(getMonitorLogNS(), descriptor.getDatasourceName(), ds, getDataDescriptorRegistry(), this);
			DataSourceMonitorTask worker = new DataSourceMonitorTask(descriptor, ctx);
			monitors.put(descriptor.getDatasourceName(), worker);
			Thread thread = new Thread(worker, String.format("DSMonitor[%s]", descriptor.getDatasourceName()));
			thread.start();
		}
	}
	
	private void stopAndRemoveDataSourceMonitor(DataSourceMonitorDescriptor descriptor) {
		synchronized (this) {
			DataSourceMonitorTask worker = monitors.remove(descriptor.getDatasourceName());
			if (worker != null) worker.stop();
		}
	}


	@Override
	protected void beforeRemoveDescriptor(DataSourceMonitorDescriptor descriptor) {
		stopAndRemoveDataSourceMonitor(descriptor);
		try {
			unDeployMonitorTable(getDataSourceFactory().getDataSource(descriptor.getDatasourceName()));
		} catch (Throwable e) {
			Log.warn(new EdgeException(descriptor, String.format("移除数据源(%s)的监听时发生错误: %s，此错误已被忽略！", descriptor.getDatasourceName(), e.getMessage())));
		}
	}

	@Override
	protected void postRemoveDescriptor(DataSourceMonitorDescriptor descriptor) {}

	@Override
	protected void beforeLoadConfiguredDescriptor(
			DataSourceMonitorDescriptor descriptor) {
		createAndStartDataSourceMonitor(descriptor);
	}
	
	@Override
	protected void beforeUnLoadConfiguredDescriptor(
			DataSourceMonitorDescriptor descriptor) {
		stopAndRemoveDataSourceMonitor(descriptor);
	}
	
	@Override
	protected void setupDescriptorIdentifier(DataSourceMonitorDescriptor descriptor, String id) {
		descriptor.setDatasourceName(id);
	}
	
	protected void unDeployMonitorTable(final DataSource ds) {
		if (isMonitorTableCreated(ds)) {
			final String[] unDeployTableSQLs = dsmSQLProviderFactory.getDsmSQLProvider(ds).getCreateMonitorLogSQLs(monitorLogTable);
			DbUtils.executeInTransaction(ds, TransactionAttribute.PROPAGATION_REQUIRES_NEW, new ExecutionCallback() {
				public void callback() {
					JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
					jdbcHelper.batchUpdate(unDeployTableSQLs);
				}
			});
		}
	}
	
	protected void deployMonitorTable(final DataSource ds) {
		if (!isMonitorTableCreated(ds)) {
			final String[] deployTableSQLs = dsmSQLProviderFactory.getDsmSQLProvider(ds).getDropMonitorLogSQLs(monitorLogTable);;
			DbUtils.executeInTransaction(ds, TransactionAttribute.PROPAGATION_REQUIRES_NEW, new ExecutionCallback() {
				public void callback() {
					JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
					jdbcHelper.batchUpdate(deployTableSQLs);
				}
			});
		}
	}

	@Override
	protected Properties beforeAddDescriptor(DataSourceMonitorDescriptor descriptor) {
		Properties props = new Properties();
		props.setProperty("datasourceName", descriptor.getDatasourceName());
		props.setProperty("monitorInterval", Long.toString(descriptor.getMonitorInterval()));
		props.setProperty("retrieveCount", Integer.toString(descriptor.getRetrieveCount()));
		props.setProperty("delRetMessage", Boolean.toString(descriptor.isRemoveAfterRetrieved()));
		props.setProperty("startMode", descriptor.getStartMode().name());
		props.setProperty("title", descriptor.getTitle());
		props.setProperty("description", descriptor.getDescription());
		props.setProperty("ordinal", Integer.valueOf(descriptor.getOrdinal()).toString());

		deployMonitorTable(getDataSourceFactory().getDataSource(descriptor.getDatasourceName()));
		return props;
	}
	
	@Override
	protected void postAddDescriptor(DataSourceMonitorDescriptor descriptor) {
		createAndStartDataSourceMonitor(descriptor);
	}

	@Override
	protected void validateDescriptor(DataSourceMonitorDescriptor descriptor) {
		Assert.notNull(descriptor.getDatasourceName(), "必须设置数据源名称。");
		Assert.isTrue(descriptor.getMonitorInterval() > 0, "监听间隔必须大于0秒。");
		Assert.isTrue(descriptor.getRetrieveCount() > 0, "每次获取消息数据必须大于0个。");
		Assert.isTrue(getDataSourceFactory().hasDataSource(descriptor.getDatasourceName()), 
				String.format("设置的数据源(%s)不存在。", descriptor.getDatasourceName()));
	}

	@Override
	protected String resolveDescriptorIdentifier(DataSourceMonitorDescriptor descriptor) {
		return descriptor.getDatasourceName();
	}
	
	@Override
	public void initContext() {
		Assert.notNull(getDataSourceFactory(), "必须设置数据源工厂参数。");
		Assert.notNull(getMessageProcessor(), "必须设置消息处理器。");
		Assert.notNull(getDataTransformer(), "必须设置数据转换器。");
		Assert.notNull(getDataBroadcaster(), "必须设置数据广播器。");
		super.initContext();
	}
	
	
	//implements DataSourceMonitorManager
	public void addMonitor(DataSourceMonitorDescriptor info) {
		addDescriptor(info);
	}

	public void removeMonitor(String dsName) {
		removeDescriptor(dsName);
	}

	public boolean isMonitored(String dsName) {
		return hasDescriptor(dsName);
	}
	
	//implements DataIntegrateProcessor
	public boolean supportData(Class<? extends DataEdgeDescriptor> descriptorClazz) {
		return ClassUtils.isAssignable(TableDataEdgeDescriptor.class, descriptorClazz);
	}

	public boolean supportData(DataEdgeDescriptor dataDescriptor) {
		return supportData((Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor));
	}
	
	/**
	 * @param batch {@link DataIntegrateBatch#getInputParameter()}必须传入一个对象数组（共两个元素）<br/>
	 * 其中{@link DataIntegrateBatch#getInputParameter()}[0]必须放置{@link DataAction}类型。<br/>
	 * 其中{@link DataIntegrateBatch#getInputParameter()}[1]必须放置对象数据或者列表。
	 */
	public void process(DataIntegrateBatch batch) {
		Assert.notNull(batch);
		Assert.notNull(batch.getDataDescriptor());
		Assert.notNull(batch.getInputParameter());
		Assert.isTrue(supportData(batch.getDataDescriptor()));
		try {
			List records = null;
			if (ClassUtils.isAssignableValue(List.class, batch.getInputParameter())) {
				records = (List<Object>) batch.getInputParameter();
			} else if (batch.getInputParameter().getClass().isArray()){
				records = Arrays.asList(ObjectUtils.toObjectArray(batch.getInputParameter()));
			}
			Assert.isTrue(records.size() > 0 && records.size() <= 2, "处理参数错误(仅允许1个或两个)。");
			DataAction dataAction = DataAction.INSERT;
			if (records.size()>1) {
				Assert.isInstanceOf(DataAction.class, records.get(0), "处理参数错误，第一个参数必须为DataAction类型。");
				dataAction = (DataAction) records.get(0);
				Object recordParameter = records.get(1);
				if (ClassUtils.isAssignableValue(List.class, recordParameter)) {
					records = (List<Object>) recordParameter;
				} else if (batch.getInputParameter().getClass().isArray()){
					records = Arrays.asList(ObjectUtils.toObjectArray(recordParameter));
				}
			}
			
			Assert.notEmpty(records, "处理的记录数必须大于0。");
			Assert.isInstanceOf(Map.class, records.get(0), "记录必须为Map类型。");
			MessageEntity msgEntity = getMessageProcessor().newMessage(null, batch.getDataDescriptor().getMessageCode(), dataAction, records, null);
			//TODO 需要实现根据描述符判断转码格式
			byte[] msgBytes = ((TableDataEdgeDescriptor)batch.getDataDescriptor()).isConversion() ? getDataTransformer().transform(batch.getDataDescriptor(), msgEntity) :
				getMessageProcessor().toMessageBytes(msgEntity, MessageFormat.XML).toString().getBytes("utf-8"); 
			getDataBroadcaster().broadcast(batch.getDataDescriptor(), msgBytes);
		} catch (Throwable e) {
			batch.setException(new EdgeException(batch.getDataDescriptor(), e.getMessage(), e));
		}
	}
	
	@Override
	public Set<Class<? extends EdgeDescriptor>> getDependClasses() {
		return dependsClasses;
	}

	@Override
	public boolean dependDescriptor(EdgeDescriptor descriptor) {
		Assert.notNull(descriptor);
		return getDependClasses().contains(ClassUtils.getUserClass(descriptor.getClass())) ?
			isMonitored(((DataSourceDescriptor)descriptor).getName()) : false;
	}
	
	@Override
	public DataSourceMonitorDescriptor[] getDescriptorByDepend(EdgeDescriptor descriptor) {
		DataSourceMonitorDescriptor returnMonitor = null;
		if (dependDescriptor(descriptor)) {
			returnMonitor = getDescriptor(((DataSourceDescriptor)descriptor).getName());
		}
		return returnMonitor == null ? new DataSourceMonitorDescriptor[0] : new DataSourceMonitorDescriptor[]{returnMonitor};
	}

	protected boolean isMonitorTableCreated(final DataSource ds) {
		Assert.notNull(ds);
		return DbUtils.executeInReadonly(ds, new ReturnExecutionCallback<Boolean>() {
			public Boolean callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				try {
					jdbcHelper.queryForInt("select count(*) from SCDM_MESSAGE");
					jdbcHelper.queryForInt("select count(*) from SCDM_MESSAGE_BAK");
					jdbcHelper.queryForInt("select count(*) from SCDM_DEFTB");
				} catch (Throwable e) {
					return false;
				}
				return true;
			}
		});
	}


	static class DataSourceMonitorContextImpl implements DataSourceMonitorContext {
		
		private final String dsName;
		
		private final DataSource dataSource;
		
		private final String systemCode;
		
		private String table;
						
		private final TableDataDescriptorRegistry<TableDataEdgeDescriptor> dataDescriptorRegistry;
		
		private final DataIntegrateProcessor dataIntegrateProcessor;
		
		DataSourceMonitorContextImpl(String systemCode, String dsName, DataSource dataSource, TableDataDescriptorRegistry<TableDataEdgeDescriptor> dataDescriptorRegsitry, DataIntegrateProcessor dataIntegrateProcessor) {
			this.dsName = dsName;
			this.dataSource = dataSource;
			this.systemCode = systemCode;
			this.dataDescriptorRegistry = dataDescriptorRegsitry;
			this.dataIntegrateProcessor = dataIntegrateProcessor;
		}
		
		public String getMonitorLogTable() {
			return table;
		}

		public DataSource getDataSource() {
			return dataSource;
		}

		public String getMonitorLogNS() {
			return systemCode;
		}


		public TableDataEdgeDescriptor[] getEdgeDescriptor(String schemaName, String tableName) {
			return dataDescriptorRegistry.getDescriptors(dsName, schemaName, tableName);
		}

		public void processMonitorData(TableDataEdgeDescriptor descriptor, DataAction dataAction, List<Map<DataProperty, String>> changedRows) {
			Assert.notNull(descriptor);
			if (CollectionUtils.isEmpty(changedRows)) return;
			dataAction = dataAction == null ? DataAction.INSERT : dataAction;
			BaseDataIntegrateBatch batch = new BaseDataIntegrateBatch(descriptor, new Object[]{dataAction, changedRows});
			dataIntegrateProcessor.process(batch);
			if (batch.hasException()) {
				Log.error(batch.getException());
			}
		}
	}


}
